package net.self.kafka.consumer;

import org.apache.ibatis.session.SqlSession;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Transaction;

import java.util.Collection;

/**
 * Re-balancer for any subscription changes.
 */
public class MyConsumerRebalancerListener implements org.apache.kafka.clients.consumer.ConsumerRebalanceListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(MyConsumerRebalancerListener.class);
    private OffsetManager offsetManager = new OffsetManager();
    private Consumer<String, String> consumer;
    private JedisPool jedisPool;
    private SqlSession sqlSession;

    public MyConsumerRebalancerListener(Consumer<String, String> consumer) {
        this.consumer = consumer;
        String redis = System.getenv("REDIS");
        String redis_password = System.getenv("REDIS_PASSWORD");
        LOGGER.info("=====start consumer redis_url:"+redis );
        LOGGER.info("=====start consumer redis_password:"+redis_password );
        if(redis==null && redis_password==null){
            jedisPool = new JedisPool(new JedisPoolConfig(), "localhost",6379,60);
        }else{
            jedisPool = new JedisPool(new JedisPoolConfig(), redis,6379,60,redis_password);
        }
        try{
//            this.sqlSession = DbAccess.getSqlSession();
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    }

    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Jedis jedis = jedisPool.getResource();
        Transaction t = jedis.multi();
        for (TopicPartition partition : partitions) {
            offsetManager.saveOffsetInExternalStore(partition.topic(), partition.partition(), consumer.position(partition), t);
        }
        jedisPool.returnResource(jedis);
    }

    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        Jedis jedis = jedisPool.getResource();
        for (TopicPartition partition : partitions) {
            consumer.seek(partition, offsetManager.readOffsetFromExternalStore(partition.topic(), partition.partition(), jedis));
        }
        jedisPool.returnResource(jedis);
    }


}
